

package etcdserver

import (
	"context"
	"encoding/json"
	"expvar"
	"fmt"
	"go.etcd.io/etcd/raft"
	"math"
	"math/rand"
	"net/http"
	"path"
	"regexp"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"github.com/coreos/go-semver/semver"
	humanize "github.com/dustin/go-humanize"
	"github.com/prometheus/client_golang/prometheus"
	"go.etcd.io/etcd/pkg/v3/notify"
	"go.etcd.io/etcd/server/v3/config"
	"go.uber.org/zap"

	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
	"go.etcd.io/etcd/api/v3/membershippb"
	"go.etcd.io/etcd/api/v3/version"
	"go.etcd.io/etcd/client/pkg/v3/fileutil"
	"go.etcd.io/etcd/client/pkg/v3/types"
	"go.etcd.io/etcd/pkg/v3/idutil"
	"go.etcd.io/etcd/pkg/v3/pbutil"
	"go.etcd.io/etcd/pkg/v3/runtime"
	"go.etcd.io/etcd/pkg/v3/schedule"
	"go.etcd.io/etcd/pkg/v3/traceutil"
	"go.etcd.io/etcd/pkg/v3/wait"
	"go.etcd.io/etcd/raft/v3"
	"go.etcd.io/etcd/raft/v3/raftpb"
	"go.etcd.io/etcd/server/v3/auth"
	"go.etcd.io/etcd/server/v3/etcdserver/api"
	"go.etcd.io/etcd/server/v3/etcdserver/api/membership"
	"go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
	"go.etcd.io/etcd/server/v3/etcdserver/api/snap"
	"go.etcd.io/etcd/server/v3/etcdserver/api/v2http/httptypes"
	stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"
	"go.etcd.io/etcd/server/v3/etcdserver/api/v2store"
	"go.etcd.io/etcd/server/v3/etcdserver/api/v3alarm"
	"go.etcd.io/etcd/server/v3/etcdserver/api/v3compactor"
	"go.etcd.io/etcd/server/v3/etcdserver/cindex"
	serverversion "go.etcd.io/etcd/server/v3/etcdserver/version"
	"go.etcd.io/etcd/server/v3/lease"
	"go.etcd.io/etcd/server/v3/lease/leasehttp"
	serverstorage "go.etcd.io/etcd/server/v3/storage"
	"go.etcd.io/etcd/server/v3/storage/backend"
	"go.etcd.io/etcd/server/v3/storage/mvcc"
	"go.etcd.io/etcd/server/v3/storage/schema"
)

const (
	DefaultSnapshotCount = 100000

	// DefaultSnapshotCatchUpEntries是慢速跟随者压缩raft存储条目后要追赶的条目数
	// 。
	// 我们希望跟随者与引导者之间有毫秒级的延迟。
	// 最大吞吐量约为10K。保持5公里的参赛成绩足以帮助
	// 追随者迎头赶上。
	DefaultSnapshotCatchUpEntries uint64 = 5000

	StoreClusterPrefix = "/0"
	StoreKeysPrefix    = "/1"

	// HealthInterval是在接受添加成员请求之前，群集应处于健康状态的最短时间。
	HealthInterval = 5 * time.Second

	purgeFileInterval = 30 * time.Second

	// etcdserver允许具有
	// 此数字对于具有5台机器的大多数群集来说已经足够了。
	maxInFlightMsgSnap = 16

	releaseDelayAfterSnapshot = 30 * time.Second

	// maxPendingRevokes是未完成的过期租约撤销的最大数量。
	maxPendingRevokes = 16

	recommendedMaxRequestBytes = 10 * 1024 * 1024

	readyPercent = 0.9

	DowngradeEnabledPath = "/downgrade/enabled"
)

var (
	// monitorVersionInterval应小于连接上的超时
	// 。或者我们将无法重新使用
	// 连接（因为它将超时）。
	monitorVersionInterval = rafthttp.ConnWriteTimeout - time.Second

	recommendedMaxRequestBytesString = humanize.Bytes(uint64(recommendedMaxRequestBytes))
	storeMemberAttributeRegexp       = regexp.MustCompile(path.Join(membership.StoreMembersPrefix, "[[:xdigit:]]{1,16}", "attributes"))
)

func init() {
	rand.Seed(time.Now().UnixNano())

	expvar.Publish(
		"file_descriptor_limit",
		expvar.Func(
			func() interface{} {
				n, _ := runtime.FDLimit()
				return n
			},
		),
	)
}

type Response struct {
	Term    uint64
	Index   uint64
	Event   *v2store.Event
	Watcher v2store.Watcher
	Err     error
}

type ServerV2 interface {
	Server
	// 获取当前集群中的 Leader 的 ID
	Leader() types.ID
	// 处理 Client 请求
	Do(ctx context.Context, r pb.Request) (Response, error)
	stats.Stats
	ClientCertAuthEnabled() bool
}

type ServerV3 interface {
	Server
	RaftStatusGetter
}

func (s *EtcdServer) ClientCertAuthEnabled() bool { return s.Cfg.ClientCertAuthEnabled }

type Server interface {
	// 向当前 etcd 集群中添加一个节点
	AddMember(ctx context.Context, memb membership.Member) ([]*membership.Member, error)
	// 从当前 etcd 集群中删除一个节点
	RemoveMember(ctx context.Context, id uint64) ([]*membership.Member, error)
	// 修改集群成员属性，如果成员 ID 不存在则返回错误
	UpdateMember(ctx context.Context, updateMemb membership.Member) ([]*membership.Member, error)
	PromoteMember(ctx context.Context, id uint64) ([]*membership.Member, error)
	ClusterVersion() *semver.Version
	Cluster() api.Cluster
	Alarms() []*pb.AlarmMember
	LeaderChangedNotify() <-chan struct{}
}

// EtcdServer是服务器接口的生产实现
type EtcdServer struct {
	// 当前已发送出去但未收到响应的快照个数。
	inflightSnapshots int64  // 必须使用原子操作才能访问；保持64位对齐。
	// 当前节点 己应用的 Entry 记录的最大索引值。
	appliedIndex      uint64 // 必须使用原子操作才能访问；保持64位对齐。
	// 当前己提交的 En位y 记录的索引值。
	committedIndex    uint64 // 必须使用原子操作才能访问；保持64位对齐。
	term              uint64 // 必须使用原子操作才能访问；保持64位对齐。
	lead              uint64 // 必须使用原子操作进行访问；保持64位对齐。

	consistIndex cindex.ConsistentIndexer // consistIndex用于获取/设置/保存consistentIndex
	// etcdserver.raftNode，它是 EtcdServer 实例与底层 etcd-raft 模块通信的桥梁。
	r            raftNode                 // 使用64位原子；保持64位对齐。
	// 当前节点将自身的信息推送到集群中其他节点之后，会将该通道关闭，
	// 也作为当前 EtcdSe凹er 实例，可以对外提供服务的一个信号。
	readych chan struct{}
	// 封装了配置信息
	Cfg     config.ServerConfig

	lgMu *sync.RWMutex
	lg   *zap.Logger
	// Wait 主要负责协调多个后台 goroutine 之间的执行。在 Wait 实例
	// 中维护了 一个 map ( map[uint64]chan interface ｛｝ 类型〉，我们可以通过 Wait.Register(id
	// uint64 ）为指定的 ID 创建一个对应的通道， ID 与通道的映射关系会记录在上述 map 中。
	// 之后，可以通过 Trigger(id uint64, x interface ｛｝） 方法将参数 x 写入 id 对应的通道中，其
	// 他监听该通道的 goroutine 就可以获取该参数 L
	w wait.Wait

	readMu sync.RWMutex
	// 主要用来协调 Linearizable Read 相关的 goroutine
	readwaitc chan struct{}
	// 主要用来协调 Linearizable Read 相关的 goroutine
	readNotifier *notifier

	// stop 、 stopping 、 done ( chan struct ｛｝类型）：
	// 在 EtcdServer.s t art （） 方法中会启动多个后
	// 台 goroutine ，其中一个后台 goroutine 会执行 EtcdServer.run （）方法，监听 stop 通道。在
	// EtcdServer. Stop （）方法中会将 stop 通道关闭，触发该 run goroutine 的结束 。在 run
	// go routine 结束之前还会将关闭 stopping 和 don e 通道，从而触发其他后台 goroutine 的
	// 关闭。

	// 停止运行GOROUTE应该关机。
	stop chan struct{}
	// 关闭时运行goroutine关闭停止。
	stopping chan struct{}
	// 当start（）中的所有goroutine完成时，done关闭。
	done chan struct{}


	// leaderChanged用于通知线性化读取循环丢弃旧的读取请求。
	leaderChanged *notify.Notifier

	errorc     chan error
	// 记录当前节点的 ID 。
	id         types.ID
	// 记录当前节点的名称及接收集群中其他节点请求的 URL 地址 。
	attributes membership.Attributes
	// 记录当前集群中全部节点的信息。
	cluster *membership.RaftCluster
	// etcd v2 版本存储。
	v2store     v2store.Store
	// 用来读写快照文件
	snapshotter *snap.Snapshotter
	// Appli erv2 接口主要功能是应用 v2 版本的 Entry 记录，其底层封装了前面介绍的 v2 存储。
	applyV2 ApplierV2

	// applierV3 接口主要功能是应用 v3 版本的 Entry 记录，其底层封装了前面介绍的 v3 存储。
	applyV3 applierV3
	// applyV3Base是没有auth或配额的核心应用程序
	applyV3Base applierV3
	// applyV3Internal是内部请求的应用程序
	applyV3Internal applierV3Internal
	// WaitTime 是在上面介绍的 Wait 之上的一层扩展。
	// 在 WaitTime 中记录的 ID 是有序的，我们可以通过 WaitTime . Wait（）方法创建指定 ID 与
	// 通道之间的映射关系 。 WaitTime.Trigger（）方法会将小于指定值的 ID 对应的通道关闭，
	// 这样就可 以通知监听相应通道的 goroutine 了 。
	applyWait       wait.WaitTime
	// etcd v3 版本的存储
	kv         mvcc.WatchableKV
	lessor     lease.Lessor
	bemu       sync.Mutex
	// v3 版本的后端存储，
	be         backend.Backend
	beHooks    *serverstorage.BackendHooks
	// 在 backend .B ackend （即这里的 be 字段）之上封装的一层存储，用于记录权限控制相关的信息，
	authStore  auth.AuthStore
	// 在 b ackend .Backend 之上封装的一层存储，用于记录报警相关的信息
	alarmStore *v3alarm.AlarmStore

	stats  *stats.ServerStats
	lstats *stats.LeaderStats
	// 用来控制 Leader 节点定期发送 SYNC 消息的频率。
	SyncTicker *time.Ticker
	// Leader 节点会对存储进行定期压缩，该字段用于控制定期压缩的频率。
	compactor v3compactor.Compactor

	// peerRt用于向对等方发送请求（版本、租约）。
	peerRt   http.RoundTripper
	// 用 于生成请求的唯一标识。
	reqIDGen *idutil.Generator

	// wgMu在服务器停止时阻止并发waitgroup变异
	wgMu sync.RWMutex
	// 在 EtcdServer.Stop （） 方法中会通过该字段等待所有的后台 goroutine 全部退出 。
	wg sync.WaitGroup

	// ctx用于etcd启动的请求，这些请求可能需要在etcd服务器关闭时取消。
	ctx    context.Context
	cancel context.CancelFunc

	leadTimeMu      sync.RWMutex
	// 记录当前节点最近一次转换成 Leader 状态 的时间戳。
	leadElectedTime time.Time

	firstCommitInTerm     *notify.Notifier
	clusterVersionChanged *notify.Notifier

	*AccessController
}

// NewServer（） 函数中会完成 EtcdServer 的初始化，也是 etcd 服务端生命周期的起始。
// ( 1 ）定义初始化过程中使用的变量，创建当前节点使用的目录
// ( 2 ）根据配置项初始化 etcd-ra企 模块使用 到的相关组件 ，例如，检测 当前 wal 目录下是否
//       存在 WAL 日志文件、初始化 v2 存储、 查找 BoltDB 数据库文件、 创建 Backend 实例 、创建
//        RoundTripper 实例等
// (3 ）根据前面对 WAL 日志文件的查找结果及当前节点启动时的配置信息，初始化 etcd-raft 模块中的 Node 实例，
func NewServer(cfg config.ServerConfig) (srv *EtcdServer, err error) {
	b, err := bootstrap(cfg)
	if err != nil {
		return nil, err
	}

	defer func() {
		if err != nil {
			b.Close()
		}
	}()

	sstats := stats.NewServerStats(cfg.Name, b.cluster.cl.String())
	lstats := stats.NewLeaderStats(cfg.Logger, b.cluster.nodeID.String())

	// (4 ）创建 EtcdServer 实例，并初始化其各个字段
	heartbeat := time.Duration(cfg.TickMs) * time.Millisecond
	srv = &EtcdServer{	// 创建 EtcdServer 实例
		readych:               make(chan struct{}),
		Cfg:                   cfg,
		lgMu:                  new(sync.RWMutex),
		lg:                    cfg.Logger,
		errorc:                make(chan error, 1),
		v2store:               b.storage.st,
		snapshotter:           b.ss,
		r:                     *b.raft.newRaftNode(b.ss, b.storage.wal.w, b.cluster.cl),
		id:                    b.cluster.nodeID,
		attributes:            membership.Attributes{Name: cfg.Name, ClientURLs: cfg.ClientURLs.StringSlice()},
		cluster:               b.cluster.cl,
		stats:                 sstats,
		lstats:                lstats,
		SyncTicker:            time.NewTicker(500 * time.Millisecond),
		peerRt:                b.prt,
		reqIDGen:              idutil.NewGenerator(uint16(b.cluster.nodeID), time.Now()),
		AccessController:      &AccessController{CORS: cfg.CORS, HostWhitelist: cfg.HostWhitelist},
		consistIndex:          b.storage.backend.ci,
		firstCommitInTerm:     notify.NewNotifier(),
		clusterVersionChanged: notify.NewNotifier(),
	}
	serverID.With(prometheus.Labels{"server_id": b.cluster.nodeID.String()}).Set(1)
	srv.cluster.SetVersionChangedNotifier(srv.clusterVersionChanged)
	// 初始化， EtcdServer.applyV2 字段
	srv.applyV2 = NewApplierV2(cfg.Logger, srv.v2store, srv.cluster)

	srv.be = b.storage.backend.be	// 初始化 EtcdServer.be 字段
	srv.beHooks = b.storage.backend.beHooks
	minTTL := time.Duration((3*cfg.ElectionTicks)/2) * heartbeat

	// 因为在 store.restore（）方法中除了恢复内存索引 ，还会重新绑定键值对与对应的 Lease ,
	// 所以需要先恢复 EtcdServer.lessor ，再恢复 EtcdServer.kv 字段
	srv.lessor = lease.NewLessor(srv.Logger(), srv.be, srv.cluster, lease.LessorConfig{
		MinLeaseTTL:                int64(math.Ceil(minTTL.Seconds())),
		CheckpointInterval:         cfg.LeaseCheckpointInterval,
		CheckpointPersist:          cfg.LeaseCheckpointPersist,
		ExpiredLeasesRetryInterval: srv.Cfg.ReqTimeout(),
	})

	tp, err := auth.NewTokenProvider(cfg.Logger, cfg.AuthToken,
		func(index uint64) <-chan struct{} {
			return srv.applyWait.Wait(index)
		},
		time.Duration(cfg.TokenTTL)*time.Second,
	)
	if err != nil {
		cfg.Logger.Warn("failed to create token provider", zap.Error(err))
		return nil, err
	}

	mvccStoreConfig := mvcc.StoreConfig{
		CompactionBatchLimit:    cfg.CompactionBatchLimit,
		CompactionSleepInterval: cfg.CompactionSleepInterval,
	}
	srv.kv = mvcc.New(srv.Logger(), srv.be, srv.lessor, mvccStoreConfig)
	// 初始化 EtcdServer.authStore 字段
	srv.authStore = auth.NewAuthStore(srv.Logger(), schema.NewAuthBackend(srv.Logger(), srv.be), tp, int(cfg.BcryptCost))

	newSrv := srv // 如果srv返回为nil，则延迟中的srv==nil 
	defer func() {
		// 在不首先关闭kv的情况下关闭后端可能会导致
		// 恢复的压缩失败，并出现关闭的发送错误
		if err != nil {
			newSrv.kv.Close()
		}
	}()
	if num := cfg.AutoCompactionRetention; num != 0 {	// 启动后台 goroutine ，进行自动压缩
		srv.compactor, err = v3compactor.New(cfg.Logger, cfg.AutoCompactionMode, num, srv.kv, srv)
		if err != nil {
			return nil, err
		}
		srv.compactor.Run()
	}
	// 初始化 applyV3Base 字段
	srv.applyV3Base = srv.newApplierV3Backend()
	srv.applyV3Internal = srv.newApplierV3Internal()
	// 初始化 alarmStore 字段以及 applyV3 字段
	if err = srv.restoreAlarms(); err != nil {
		return nil, err
	}

	if srv.Cfg.EnableLeaseCheckpoint {
		// 设置检查点启用租约检查点功能。
		srv.lessor.SetCheckpointer(func(ctx context.Context, cp *pb.LeaseCheckpointRequest) {
			srv.raftRequestOnce(ctx, pb.InternalRaftRequest{LeaseCheckpoint: cp})
		})
	}

	// 创建 rafthttp.Transport 实例
	// TODO:将传输初始化移到远程
	tr := &rafthttp.Transport{
		Logger:      cfg.Logger,
		TLSInfo:     cfg.PeerTLSInfo,
		DialTimeout: cfg.PeerDialTimeout(),
		ID:          b.cluster.nodeID,
		URLs:        cfg.PeerURLs,
		ClusterID:   b.cluster.cl.ID(),
		// 需要注意，在这里传入的 rafthttp . Raft 接口实现是 EtcdServer 实例 。 EtcdServer
		// 对 Raft 接口 的实现比较简单，它会直接将调用委托给底层的 raftNode 实例 。
		Raft:        srv,
		Snapshotter: b.ss,
		ServerStats: sstats,
		LeaderStats: lstats,
		ErrorC:      srv.errorc,
	}
	// 启动 rafthttp.Transport 实例
	if err = tr.Start(); err != nil {
		return nil, err
	}
	// 向 rafthttp.Transport 实例 中 添加集群中各个节点对应的 Peer 实例和 Remote 实例
	for _, m := range b.cluster.remotes {
		if m.ID != b.cluster.nodeID {
			tr.AddRemote(m.ID, m.PeerURLs)
		}
	}
	for _, m := range b.cluster.cl.Members() {
		if m.ID != b.cluster.nodeID {
			tr.AddPeer(m.ID, m.PeerURLs)
		}
	}
	srv.r.transport = tr	// 设置 raft.Node.transport 字段

	return srv, nil
}

func (s *EtcdServer) Logger() *zap.Logger {
	s.lgMu.RLock()
	l := s.lg
	s.lgMu.RUnlock()
	return l
}

func tickToDur(ticks int, tickMs uint) string {
	return fmt.Sprintf("%v", time.Duration(ticks)*time.Duration(tickMs)*time.Millisecond)
}

func (s *EtcdServer) adjustTicks() {
	lg := s.Logger()
	clusterN := len(s.cluster.Members())

	// 单节点重新启动，或单节点从快照恢复
	if clusterN == 1 {
		ticks := s.Cfg.ElectionTicks - 1
		lg.Info(
			"started as single-node; fast-forwarding election ticks",
			zap.String("local-member-id", s.ID().String()),
			zap.Int("forward-ticks", ticks),
			zap.String("forward-duration", tickToDur(ticks, s.Cfg.TickMs)),
			zap.Int("election-ticks", s.Cfg.ElectionTicks),
			zap.String("election-timeout", tickToDur(s.Cfg.ElectionTicks, s.Cfg.TickMs)),
		)
		s.r.advanceTicks(ticks)
		return
	}

	if !s.Cfg.InitialElectionTickAdvance {
		lg.Info("skipping initial election tick advance", zap.Int("election-ticks", s.Cfg.ElectionTicks))
		return
	}
	lg.Info("starting initial election tick advance", zap.Int("election-ticks", s.Cfg.ElectionTicks))

	// 重试至“rafthttp.ConnReadTimeout”，这是5秒
	// 直到对等连接报告；否则：
	// 1。所有连接失败，或
	// 2。无活动对等点，或
	// 3。在没有快照的情况下重新启动单节点
	// 然后，什么也不做，因为提前勾号不会产生任何影响
	waitTime := rafthttp.ConnReadTimeout
	itv := 50 * time.Millisecond
	for i := int64(0); i < int64(waitTime/itv); i++ {
		select {
		case <-time.After(itv):
		case <-s.stopping:
			return
		}

		peerN := s.r.transport.ActivePeers()
		if peerN > 1 {
			// 多节点收到对等连接报告
			// 调整勾号，如果慢速引导消息接收
			ticks := s.Cfg.ElectionTicks - 2

			lg.Info(
				"initialized peer connections; fast-forwarding election ticks",
				zap.String("local-member-id", s.ID().String()),
				zap.Int("forward-ticks", ticks),
				zap.String("forward-duration", tickToDur(ticks, s.Cfg.TickMs)),
				zap.Int("election-ticks", s.Cfg.ElectionTicks),
				zap.String("election-timeout", tickToDur(s.Cfg.ElectionTicks, s.Cfg.TickMs)),
				zap.Int("active-remote-members", peerN),
			)

			s.r.advanceTicks(ticks)
			return
		}
	}
}

// Start对服务器执行任何必要的初始化，以使其
// 开始服务请求。必须在执行或处理之前调用它。
// 启动必须是非阻塞的；任何长时间运行的服务器功能
// 都应该在goroutines中实现。
// 执行完 EtcdServer.NewServer（）、注册 Handler 等初始化操作之后， 下面会紧接着调用
// EtcdServer.Start （） 方法启动当前节点， 对外提供服务。
func (s *EtcdServer) Start() {
	// 其中会启动一个后台 goroutine， 执行 EtcdServer.run （ ）方法
	s.start()
	s.GoAttach(func() { s.adjustTicks() })
	// 启动一个后台 goroutine ，将 当前节点的相关信息发送到 集群其他节点
	s.GoAttach(func() { s.publish(s.Cfg.ReqTimeout()) })
	// 启动一个后台 goroutine ，定义清理 WAL 日志文件和快照文件
	s.GoAttach(s.purgeFile)
	// 启动 一个后台 goroutine ，实现一些监控相关的功能，
	s.GoAttach(func() { monitorFileDescriptor(s.Logger(), s.stopping) })
	// 启动一个后台 goroutine 监控集群中其他节点的版本信息，主妾是在版本升级的时候使用，
	s.GoAttach(s.monitorClusterVersions)
	s.GoAttach(s.monitorStorageVersion)
	// 启动一个后台 goroutine ，用来实现 Linearizable Read 的功能
	s.GoAttach(s.linearizableReadLoop)
	s.GoAttach(s.monitorKVHash)
	s.GoAttach(s.monitorDowngrade)
}

// start在新的goroutine中准备并启动服务器。
// 在服务器发送启动后修改其字段不再安全。
// 此函数仅用于测试。
// EtcdServer.start（）方法，该方法会初始化 EtcdServer 实例中剩余的未初
// 始化字段，然后启动后台 goroutine 来执行 EtcdServer.run （） 方法
func (s *EtcdServer) start() {
	lg := s.Logger()

	if s.Cfg.SnapshotCount == 0 {
		lg.Info(
			"updating snapshot-count to default",
			zap.Uint64("given-snapshot-count", s.Cfg.SnapshotCount),
			zap.Uint64("updated-snapshot-count", DefaultSnapshotCount),
		)
		s.Cfg.SnapshotCount = DefaultSnapshotCount
	}
	if s.Cfg.SnapshotCatchUpEntries == 0 {
		lg.Info(
			"updating snapshot catch-up entries to default",
			zap.Uint64("given-snapshot-catchup-entries", s.Cfg.SnapshotCatchUpEntries),
			zap.Uint64("updated-snapshot-catchup-entries", DefaultSnapshotCatchUpEntries),
		)
		s.Cfg.SnapshotCatchUpEntries = DefaultSnapshotCatchUpEntries
	}

	s.w = wait.New()
	s.applyWait = wait.NewTimeList()
	s.done = make(chan struct{})
	s.stop = make(chan struct{})
	s.stopping = make(chan struct{}, 1)
	s.ctx, s.cancel = context.WithCancel(context.Background())
	s.readwaitc = make(chan struct{}, 1)
	s.readNotifier = newNotifier()
	s.leaderChanged = notify.NewNotifier()
	if s.ClusterVersion() != nil {
		lg.Info(
			"starting etcd server",
			zap.String("local-member-id", s.ID().String()),
			zap.String("local-server-version", version.Version),
			zap.String("cluster-id", s.Cluster().ID().String()),
			zap.String("cluster-version", version.Cluster(s.ClusterVersion().String())),
		)
		membership.ClusterVersionMetrics.With(prometheus.Labels{"cluster_version": version.Cluster(s.ClusterVersion().String())}).Set(1)
	} else {
		lg.Info(
			"starting etcd server",
			zap.String("local-member-id", s.ID().String()),
			zap.String("local-server-version", version.Version),
			zap.String("cluster-version", "to_be_decided"),
		)
	}

	// TODO:如果这是一个空日志，则将所有对等信息
	// 写入第一个条目
	go s.run()
}

// EtcdSe凹er.Start（） 方法中会启动两个后 台 goroutine ， 其中一个后台 goroutine 负责定期清
// 理 WAL 日志文件 ，另 一 个后台 goroutine 负 责定期清理快照文 件，相应的逻 辑位于
// EtcdServer.purgeFile （） 方法中
func (s *EtcdServer) purgeFile() {
	lg := s.Logger()
	var dberrc, serrc, werrc <-chan error
	var dbdonec, sdonec, wdonec <-chan struct{}
	if s.Cfg.MaxSnapFiles > 0 {
		// 这里会启动一个后台 goroutine ，定期清理快照文件（默认 purgeFileinterval 的值为 30s)
		dbdonec, dberrc = fileutil.PurgeFileWithDoneNotify(lg, s.Cfg.SnapDir(), "snap.db", s.Cfg.MaxSnapFiles, purgeFileInterval, s.stopping)
		sdonec, serrc = fileutil.PurgeFileWithDoneNotify(lg, s.Cfg.SnapDir(), "snap", s.Cfg.MaxSnapFiles, purgeFileInterval, s.stopping)
	}
	if s.Cfg.MaxWALFiles > 0 {
		// 启动一个后 台 go routine ，定期 清理 WAL 日志文件 （ 默认 purgeFileint erval 的位为 30s )
		wdonec, werrc = fileutil.PurgeFileWithDoneNotify(lg, s.Cfg.WALDir(), "wal", s.Cfg.MaxWALFiles, purgeFileInterval, s.stopping)
	}

	select {
	case e := <-dberrc:
		lg.Fatal("failed to purge snap db file", zap.Error(e))
	case e := <-serrc:
		lg.Fatal("failed to purge snap file", zap.Error(e))
	case e := <-werrc:
		lg.Fatal("failed to purge wal file", zap.Error(e))
	case <-s.stopping:
		if dbdonec != nil {
			<-dbdonec
		}
		if sdonec != nil {
			<-sdonec
		}
		if wdonec != nil {
			<-wdonec
		}
		return
	}
}

func (s *EtcdServer) Cluster() api.Cluster { return s.cluster }

func (s *EtcdServer) ApplyWait() <-chan struct{} { return s.applyWait.Wait(s.getCommittedIndex()) }

type ServerPeer interface {
	ServerV2
	RaftHandler() http.Handler
	LeaseHandler() http.Handler
}

func (s *EtcdServer) LeaseHandler() http.Handler {
	if s.lessor == nil {
		return nil
	}
	return leasehttp.NewHandler(s.lessor, s.ApplyWait)
}

func (s *EtcdServer) RaftHandler() http.Handler { return s.r.transport.Handler() }

type ServerPeerV2 interface {
	ServerPeer
	HashKVHandler() http.Handler
	DowngradeEnabledHandler() http.Handler
}

func (s *EtcdServer) DowngradeInfo() *serverversion.DowngradeInfo { return s.cluster.DowngradeInfo() }

type downgradeEnabledHandler struct {
	lg      *zap.Logger
	cluster api.Cluster
	server  *EtcdServer
}

func (s *EtcdServer) DowngradeEnabledHandler() http.Handler {
	return &downgradeEnabledHandler{
		lg:      s.Logger(),
		cluster: s.cluster,
		server:  s,
	}
}

func (h *downgradeEnabledHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		w.Header().Set("Allow", http.MethodGet)
		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
		return
	}

	w.Header().Set("X-Etcd-Cluster-ID", h.cluster.ID().String())

	if r.URL.Path != DowngradeEnabledPath {
		http.Error(w, "bad path", http.StatusBadRequest)
		return
	}

	ctx, cancel := context.WithTimeout(context.Background(), h.server.Cfg.ReqTimeout())
	defer cancel()

	// 使用线性降级信息服务
	if err := h.server.linearizableReadNotify(ctx); err != nil {
		http.Error(w, fmt.Sprintf("failed linearized read: %v", err),
			http.StatusInternalServerError)
		return
	}
	enabled := h.server.DowngradeInfo().Enabled
	w.Header().Set("Content-Type", "text/plain")
	w.Write([]byte(strconv.FormatBool(enabled)))
}

// 进程接收raft消息并将其应用于服务器的raft状态
// 机器，与给定上下文的任何超时有关。
func (s *EtcdServer) Process(ctx context.Context, m raftpb.Message) error {
	lg := s.Logger()
	if s.cluster.IsIDRemoved(types.ID(m.From)) {
		lg.Warn(
			"rejected Raft message from removed member",
			zap.String("local-member-id", s.ID().String()),
			zap.String("removed-member-id", types.ID(m.From).String()),
		)
		return httptypes.NewHTTPError(http.StatusForbidden, "cannot process message from removed member")
	}
	if m.Type == raftpb.MsgApp {
		s.stats.RecvAppendReq(types.ID(m.From).String(), m.Size())
	}
	return s.r.Step(ctx, m)
}

func (s *EtcdServer) IsIDRemoved(id uint64) bool { return s.cluster.IsIDRemoved(types.ID(id)) }

func (s *EtcdServer) ReportUnreachable(id uint64) { s.r.ReportUnreachable(id) }

// 报告快照将快照发送状态报告给raft状态机，
// 并从快照存储中清除已使用的快照。
func (s *EtcdServer) ReportSnapshot(id uint64, status raft.SnapshotStatus) {
	s.r.ReportSnapshot(id, status)
}

type etcdProgress struct {
	confState raftpb.ConfState
	snapi     uint64
	appliedt  uint64
	appliedi  uint64
}

// raftReadyHandler 的功能：在结构体 EtcdServer 中记录了当前节点的状态
// 信息，例如， 当前是否是 Leader 节点、 Entry记录的提交位置（ cornmittedlndex 字段〉 等。在
// raftNode.start（）方法处理 Ready 实例的过程中，会涉及这些信息的修改， raftReadyHandler 中封
// 装了 updateLeadership 和 updateCornmittedindex 两个回调函数，这样就可以在 raftNode 中通过这
// 两个回调函数，修改 EtcdServer 中的相应字段了。
type raftReadyHandler struct {
	getLead              func() (lead uint64)
	updateLead           func(lead uint64)
	updateLeadership     func(newLeader bool)
	updateCommittedIndex func(uint64)
}

// run（）方法是 EtcdServer 启动的核心， 其中会启动
// etcdserver.raftNode 实例， 然后处理 etcd-raft 模块返回的 Ready 实例，
func (s *EtcdServer) run() {
	lg := s.Logger()

	sn, err := s.r.raftStorage.Snapshot()
	if err != nil {
		lg.Panic("failed to get snapshot from Raft storage", zap.Error(err))
	}

	// FIFO 调度器
	// 异步接受应用数据包，按顺序分派进度
	sched := schedule.NewFIFOScheduler()

	var (
		smu   sync.RWMutex
		syncC <-chan time.Time
	)
	// setSyncC（）和 getSync（）方法是用来设置发送 SYNC 消息的定时器，这两个回调函数会
	// 与 raftReadyHandler 配合使用
	setSyncC := func(ch <-chan time.Time) {
		smu.Lock()
		syncC = ch
		smu.Unlock()
	}
	getSyncC := func() (ch <-chan time.Time) {
		smu.RLock()
		ch = syncC
		smu.RUnlock()
		return
	}
	rh := &raftReadyHandler{
		getLead:    func() (lead uint64) { return s.getLead() },
		updateLead: func(lead uint64) { s.setLead(lead) },
		// raftNode 在处理 etcd-raft 模块返回的 Ready.SoftState 字段时，会调用
		// raftReadyHandler.updateLeadership （ ）回调函数 ， 其中会根据当前节点的状态和
		// Leader 节点是否发生变化完成一些相应的操作
		updateLeadership: func(newLeader bool) {
			if !s.isLeader() {
				if s.lessor != nil {
					s.lessor.Demote()	// 调用 lessor.Demote （） 将该节点的 lessor 实例降级
				}
				if s.compactor != nil {	// 非 Leader 节点暂停自动压缩，
					s.compactor.Pause()
				}
				setSyncC(nil)	// 非 Leader 节点不会发送 SYNC 消息 ， 将该定时器设置为 nil
			} else {
				if newLeader {
					t := time.Now()
					s.leadTimeMu.Lock()
					// 如果发生 Leader 节点的切换，且当前节点成为 Leader 节点，如j 初始化
					// leadElectedTime 字段，该字段记录了当前节点最近一次成为 Leader 节点的时间
					s.leadElectedTime = t
					s.leadTimeMu.Unlock()
				}
				setSyncC(s.SyncTicker.C)	// Leader 节点会定期发送 SYNC 消息，恢复该定时器
				if s.compactor != nil {	// 重启自动压缩的功能
					s.compactor.Resume()
				}
			}
			if newLeader {
				s.leaderChanged.Notify()
			}
			// TODO:删除零检查
			// 当前测试实用程序不提供统计
			if s.stats != nil {
				s.stats.BecomeLeader()
			}
		},
		// 在 raftNode 处理 apply 实例时会调用 updateCommittedindex （）函数，该函数会根据 apply
		// 实例中封装的待应用 Entry 记录和快照数据确定当前的 committedindex 值， 然后
		// 调用 raftReadyHandler 中的同名回调函数更新 EtcdServer.committedindex 字段值
		updateCommittedIndex: func(ci uint64) {
			cci := s.getCommittedIndex()
			if ci > cci {
				s.setCommittedIndex(ci)
			}
		},
	}
	// 启动 raftNode ，其中会启动后台 goroutine 处理 etcd-raft 模块返回的 Ready 实例
	s.r.start(rh)

	ep := etcdProgress{	// ／记录当前快照相关的元数据信息和己应用 Entry 记录的位置信息
		confState: sn.Metadata.ConfState,
		snapi:     sn.Metadata.Index,
		appliedt:  sn.Metadata.Term,
		appliedi:  sn.Metadata.Index,
	}

	defer func() {
		s.wgMu.Lock() // 阻止并发等待组在停止
		close(s.stopping)
		s.wgMu.Unlock()
		s.cancel()
		sched.Stop()

		// 停止raft时在GoAttach中添加GOUTRATION以便wal保持打开
		s.wg.Wait()

		s.SyncTicker.Stop()

		// 必须停止raftafter scheduler--etcdserver可以泄漏raft HTTP管道
		// 在raft停止传输后添加对等方
		s.r.stop()

		s.Cleanup()

		close(s.done)
	}()

	var expiredLeaseC <-chan []*lease.Lease
	if s.lessor != nil {
		expiredLeaseC = s.lessor.ExpiredLeasesC()
	}

	for {
		select {
		case ap := <-s.r.apply():	// 读取 raftNode.applyc 通过中的 apply 实例并进行处理
			f := func(context.Context) { s.applyAll(&ep, &ap) }
			sched.Schedule(f)
		case leases := <-expiredLeaseC:	// 监听 expiredLeaseC 通道
			s.GoAttach(func() {	// 启动羊独的 goroutine
				// 通过并行化
				c := make(chan struct{}, maxPendingRevokes)	// 用于限流 ，上 限是 16
				for _, lease := range leases {	// 遍历过期的 Lease 实例
					select {
					case c <- struct{}{}: 	// 向 c 通过中添加一 个空结构体
					case <-s.stopping:
						return
					}
					lid := lease.ID
					s.GoAttach(func() {	// 启动一个后台线程 ， 完成指定 Lease 的撤销
						ctx := s.authStore.WithRoot(s.ctx)
						_, lerr := s.LeaseRevoke(ctx, &pb.LeaseRevokeRequest{ID: int64(lid)})
						if lerr == nil {
							leaseExpired.Inc()
						} else {
							lg.Warn(
								"failed to revoke lease",
								zap.String("lease-id", fmt.Sprintf("%016x", lid)),
								zap.Error(lerr),
							)
						}

						<-c
					})
				}
			})
		case err := <-s.errorc:
			lg.Warn("server error", zap.Error(err))
			lg.Warn("data-dir used by this member must be removed")
			return
		case <-getSyncC():	// 定时发送 SYNC 消息
			if s.v2store.HasTTLKeys() {	// 如果 v2 存储中只有永久节点，则无须发送 SYNC
				s.sync(s.Cfg.ReqTimeout())	// 发送 SYNC 消息的目的是为了清理 v2 存储中的过期节点
			}
		case <-s.stop:
			return
		}
	}
}

// 清除将删除EtcdServer分配的对象。
// /中的NewServer未调用EtcdServer:：Start（负责清理）的情况。
func (s *EtcdServer) Cleanup() {
	// 如果在没有启用v3的情况下运行
	// 或运行单元测试，则出租方和后端可以为零。
	if s.lessor != nil {
		s.lessor.Stop()
	}
	if s.kv != nil {
		s.kv.Close()
	}
	if s.authStore != nil {
		s.authStore.Close()
	}
	if s.be != nil {
		s.be.Close()
	}
	if s.compactor != nil {
		s.compactor.Stop()
	}
}

// EtcdServer.applyAll （） 方法中首先调用 applySnapshot（） 方法处理 apply 实例 中记录的快照数
// 据，然后调用 applyEntries （） 方法处理 apply 实例中的 Entry 记录，之后根据 apply 实例的处理结
// 果检测是否需要生成新的快照文件， 最后处理 MsgSnap 消息
func (s *EtcdServer) applyAll(ep *etcdProgress, apply *apply) {
	s.applySnapshot(ep, apply)	// 调用 applySnapshot （）方法处理 apply 实例中记录的快照数据
	s.applyEntries(ep, apply)	// 调用 applyEntries （）方法处理 apply 实例 中的 Entry 记录

	proposalsApplied.Set(float64(ep.appliedi))
	// etcdProgress.appliedi 记录了已应用 Entry 的索引值。 这里通过调用
	// WaitTirne.Trigger （）方法 ， 将 id 小于 etcdProgress.appliedi 的 Entry 对应的通道全部关闭，
	// 这样就可以通知其他监听远远的 goroutine ，例如， linearizableReadLoop goroutine
	s.applyWait.Trigger(ep.appliedi)

	// 当 Ready 处理基本完成时 ， 会向 notifyc 通道中 写入一个信
	//  号，通知当前 goroutine 去检测是否需要生成快照
	<-apply.notifyc

	s.triggerSnapshot(ep)	// 根据当前状态决定是否触发快照的生成
	select {
	// 在 raftNode 中处理 Ready 实例时 ，如果并没有直接发送 MsgSnap 消息 ，而是将其写入 rnsgSnapC 边
	// 道 中，这里会读取 rnsgSnapC 通过，并完成快照数据的发送
	case m := <-s.r.msgSnapC:
		// 将 v2 存储的快照数据和 v3 存储的数据合并成完整的快照、数据
		merged := s.createMergedSnapshotMessage(m, ep.appliedt, ep.appliedi, ep.confState)
		// 发送快照数据
		s.sendMergedSnap(merged)
	default:
	}
}

// EtcdServer.applySnapshot（）方法会先等待 raftNode 将快照数据持久化到磁盘中，
// 之后根据快照元数据查找 BoltDB 数据库文件并重建 Backend 实例， 最后根据重建后 的存储更
// 新本地 RaftCluster 实例 。
func (s *EtcdServer) applySnapshot(ep *etcdProgress, apply *apply) {
	// 检测待应用的快炜、数据是否为空 ， 如果为空则直接返回
	if raft.IsEmptySnap(apply.snapshot) {
		return
	}
	applySnapshotInProgress.Inc()

	lg := s.Logger()
	lg.Info(
		"applying snapshot",
		zap.Uint64("current-snapshot-index", ep.snapi),
		zap.Uint64("current-applied-index", ep.appliedi),
		zap.Uint64("incoming-leader-snapshot-index", apply.snapshot.Metadata.Index),
		zap.Uint64("incoming-leader-snapshot-term", apply.snapshot.Metadata.Term),
	)
	defer func() {
		lg.Info(
			"applied snapshot",
			zap.Uint64("current-snapshot-index", ep.snapi),
			zap.Uint64("current-applied-index", ep.appliedi),
			zap.Uint64("incoming-leader-snapshot-index", apply.snapshot.Metadata.Index),
			zap.Uint64("incoming-leader-snapshot-term", apply.snapshot.Metadata.Term),
		)
		applySnapshotInProgress.Dec()
	}()

	if apply.snapshot.Metadata.Index <= ep.appliedi {
		// 如果该快照中最后一条 Entry 的索引佳小于当前节点己应用 Entry 索引值，则程序异常结束
		lg.Panic(
			"unexpected leader snapshot from outdated index",
			zap.Uint64("current-snapshot-index", ep.snapi),
			zap.Uint64("current-applied-index", ep.appliedi),
			zap.Uint64("incoming-leader-snapshot-index", apply.snapshot.Metadata.Index),
			zap.Uint64("incoming-leader-snapshot-term", apply.snapshot.Metadata.Term),
		)
	}

	// raftNode 在将快照数据写入磁盘文件之后，会向 notifyc
	// 通道中写入一个空结构体作为信号，这里会阻塞等待该信号
	<-apply.notifyc
	// 根据快照信息查找对应的 BoltDB 数据库文件，并创建新的 Backend 实例
	newbe, err := serverstorage.OpenSnapshotBackend(s.Cfg, s.snapshotter, apply.snapshot, s.beHooks)
	if err != nil {
		lg.Panic("failed to open snapshot backend", zap.Error(err))
	}

	// 因为在 store.restore （）方法中除了恢复内存索引，还会重新绑定键值对与对应的 Lease,
	// 所以需妥允恢复 EtcdServer.lessor ，再恢复 EtcdServer.kv 字段
	if s.lessor != nil {
		lg.Info("restoring lease store")

		s.lessor.Recover(newbe, func() lease.TxnDelete { return s.kv.Write(traceutil.TODO()) })

		lg.Info("restored lease store")
	}

	lg.Info("restoring mvcc store")

	if err := s.kv.Restore(newbe); err != nil {
		lg.Panic("failed to restore mvcc store", zap.Error(err))
	}
	// 重置 EtcdServer.consistindex 字段
	s.consistIndex.SetBackend(newbe)
	lg.Info("restored mvcc store", zap.Uint64("consistent-index", s.consistIndex.ConsistentIndex()))

	// 关闭旧后端可能会阻塞，直到后端上的所有TXN 
	// 完成。
	// 我们不想等待关闭旧后端。
	s.bemu.Lock()
	oldbe := s.be
	go func() {
		lg.Info("closing old backend file")
		defer func() {
			lg.Info("closed old backend file")
		}()
		// 因为此时可能还有事务在执行，关闭旧 Backend 实例 可能会被阻塞，所以这里启动一个后
		// 台 goroutine 用来关闭 Backend 实例
		if err := oldbe.Close(); err != nil {
			lg.Panic("failed to close old backend", zap.Error(err))
		}
	}()

	s.be = newbe	// 更新 EtcdServer 实例中使用的 Backend 实例
	s.bemu.Unlock()

	lg.Info("restoring alarm store")
	// 恢复 EtcdServer 中的 alarmStore 和 authStore ，它们分别对应 BoltDB 中的 alarm Bucket 和 auth Bukcet
	if err := s.restoreAlarms(); err != nil {
		lg.Panic("failed to restore alarm store", zap.Error(err))
	}

	lg.Info("restored alarm store")

	if s.authStore != nil {
		lg.Info("restoring auth store")

		s.authStore.Recover(schema.NewAuthBackend(lg, newbe))

		lg.Info("restored auth store")
	}

	lg.Info("restoring v2 store")
	// 恢复 v2 版本存储
	if err := s.v2store.Recovery(apply.snapshot.Data); err != nil {
		lg.Panic("failed to restore v2 store", zap.Error(err))
	}

	if err := serverstorage.AssertNoV2StoreContent(lg, s.v2store, s.Cfg.V2Deprecation); err != nil {
		lg.Panic("illegal v2store content", zap.Error(err))
	}

	lg.Info("restored v2 store")

	s.cluster.SetBackend(schema.NewMembershipBackend(lg, newbe))	// 设置 RaftCluster.backend 实例

	lg.Info("restoring cluster configuration")

	s.cluster.Recover(api.UpdateCapability)	// 恢复本地的集群信息，

	lg.Info("restored cluster configuration")
	lg.Info("removing old peers from network")

	// 清空 Transport 中所有 Peer 实例，并根据恢复后的 RaftCluster 实例重新添加
	s.r.transport.RemoveAllPeers()

	lg.Info("removed old peers from network")
	lg.Info("adding peers from new cluster configuration")

	for _, m := range s.cluster.Members() {
		if m.ID == s.ID() {
			continue
		}
		s.r.transport.AddPeer(m.ID, m.PeerURLs)
	}

	lg.Info("added peers from new cluster configuration")
	// 更新 etcdProgress ，其中涉及已应用 Entry 记录的 Term 值、 Index 值和快照相关信息
	ep.appliedt = apply.snapshot.Metadata.Term
	ep.appliedi = apply.snapshot.Metadata.Index
	ep.snapi = ep.appliedi
	ep.confState = apply.snapshot.Metadata.ConfState
}

// 应用完快照数据之后 ， run goroutine 紧接着会调用 EtcdServer.applyEntries （）方法处理待应用
// 的 Entry 记录
func (s *EtcdServer) applyEntries(ep *etcdProgress, apply *apply) {
	// 检测是否存在待应用的 Entry 记录，如果为空则直接返回（
	if len(apply.entries) == 0 {
		return
	}
	firsti := apply.entries[0].Index
	if firsti > ep.appliedi+1 {	// 检测待应用的第一条 Entry 记录是否合法
		lg := s.Logger()
		lg.Panic(
			"unexpected committed entry index",
			zap.Uint64("current-applied-index", ep.appliedi),
			zap.Uint64("first-committed-entry-index", firsti),
		)
	}
	var ents []raftpb.Entry
	if ep.appliedi+1-firsti < uint64(len(apply.entries)) {
		// 忽略己应用的 Entry 记录，只保留未应用的 Entry 记录
		ents = apply.entries[ep.appliedi+1-firsti:]
	}
	if len(ents) == 0 {
		return
	}
	var shouldstop bool
	// 调用 apply （）方法应用 ents 中的 Entry 记录
	if ep.appliedt, ep.appliedi, shouldstop = s.apply(ents, &ep.confState); shouldstop {
		go s.stopWithDelay(10*100*time.Millisecond, fmt.Errorf("the member has been permanently removed from the cluster"))
	}
}

// EtcdServer.triggerSnapshot（）方法对是否需要生成新快照文件的判定
func (s *EtcdServer) triggerSnapshot(ep *etcdProgress) {
	if ep.appliedi-ep.snapi <= s.Cfg.SnapshotCount {
		return	// 连续应用一定莹的 Entry 记录， 会触发快照的生成（ snapCount 默认为 100000 条 ）
	}

	lg := s.Logger()
	lg.Info(
		"triggering snapshot",
		zap.String("local-member-id", s.ID().String()),
		zap.Uint64("local-member-applied-index", ep.appliedi),
		zap.Uint64("local-member-snapshot-index", ep.snapi),
		zap.Uint64("local-member-snapshot-count", s.Cfg.SnapshotCount),
	)

	s.snapshot(ep.appliedi, ep.confState)	// 创建新的快照文件
	ep.snapi = ep.appliedi	// 更新 etcdProgress.snapi
}

func (s *EtcdServer) hasMultipleVotingMembers() bool {
	return s.cluster != nil && len(s.cluster.VotingMemberIDs()) > 1
}

func (s *EtcdServer) isLeader() bool {
	return uint64(s.ID()) == s.Lead()
}

// MoveLeader将leader转让给给定受让人。
func (s *EtcdServer) MoveLeader(ctx context.Context, lead, transferee uint64) error {
	if !s.cluster.IsMemberExist(types.ID(transferee)) || s.cluster.Member(types.ID(transferee)).IsLearner {
		return ErrBadLeaderTransferee
	}

	now := time.Now()
	interval := time.Duration(s.Cfg.TickMs) * time.Millisecond

	lg := s.Logger()
	lg.Info(
		"leadership transfer starting",
		zap.String("local-member-id", s.ID().String()),
		zap.String("current-leader-member-id", types.ID(lead).String()),
		zap.String("transferee-member-id", types.ID(transferee).String()),
	)

	s.r.TransferLeadership(ctx, lead, transferee)
	for s.Lead() != transferee {
		select {
		case <-ctx.Done(): // 超时
			return ErrTimeoutLeaderTransfer
		case <-time.After(interval):
		}
	}

	// TODO:清空所有请求，或将所有消息丢弃到旧领导
	lg.Info(
		"leadership transfer finished",
		zap.String("local-member-id", s.ID().String()),
		zap.String("old-leader-member-id", types.ID(lead).String()),
		zap.String("new-leader-member-id", types.ID(transferee).String()),
		zap.Duration("took", time.Since(now)),
	)
	return nil
}

// 转移领导将领导转移到选定的受让人。
func (s *EtcdServer) TransferLeadership() error {
	lg := s.Logger()
	if !s.isLeader() {
		lg.Info(
			"skipped leadership transfer; local server is not leader",
			zap.String("local-member-id", s.ID().String()),
			zap.String("current-leader-member-id", types.ID(s.Lead()).String()),
		)
		return nil
	}

	if !s.hasMultipleVotingMembers() {
		lg.Info(
			"skipped leadership transfer for single voting member cluster",
			zap.String("local-member-id", s.ID().String()),
			zap.String("current-leader-member-id", types.ID(s.Lead()).String()),
		)
		return nil
	}

	transferee, ok := longestConnected(s.r.transport, s.cluster.VotingMemberIDs())
	if !ok {
		return ErrUnhealthy
	}

	tm := s.Cfg.ReqTimeout()
	ctx, cancel := context.WithTimeout(s.ctx, tm)
	err := s.MoveLeader(ctx, s.Lead(), uint64(transferee))
	cancel()
	return err
}

// 硬停止停止服务器，而不与群集中的其他成员协调。
func (s *EtcdServer) HardStop() {
	select {
	case s.stop <- struct{}{}:
	case <-s.done:
		return
	}
	<-s.done
}

// Stop优雅地停止服务器，并关闭正在运行的goroutine。
// 启动后应调用Stop，否则它将永远阻塞。
// 停止leader时，Stop会在停止服务器之前将其领导权转移到其一个对等方
// 中。
// Stop终止服务器并执行任何必要的终结。
// 调用Stop后无法调用Do和进程。
func (s *EtcdServer) Stop() {
	lg := s.Logger()
	if err := s.TransferLeadership(); err != nil {
		lg.Warn("leadership transfer failed", zap.String("local-member-id", s.ID().String()), zap.Error(err))
	}
	s.HardStop()
}

// ReadyNotify返回一个通道，当服务器
// 准备好服务客户端请求时，该通道将关闭
func (s *EtcdServer) ReadyNotify() <-chan struct{} { return s.readych }

func (s *EtcdServer) stopWithDelay(d time.Duration, err error) {
	select {
	case <-time.After(d):
	case <-s.done:
	}
	select {
	case s.errorc <- err:
	default:
	}
}

// StopNotify返回一个通道，该通道在服务器停止时接收空结构
// 。
func (s *EtcdServer) StopNotify() <-chan struct{} { return s.done }

// StoppingNotify在服务器停止时返回接收空结构
// 的通道。
func (s *EtcdServer) StoppingNotify() <-chan struct{} { return s.stopping }

func (s *EtcdServer) SelfStats() []byte { return s.stats.JSON() }

func (s *EtcdServer) LeaderStats() []byte {
	lead := s.getLead()
	if lead != uint64(s.id) {
		return nil
	}
	return s.lstats.JSON()
}

func (s *EtcdServer) StoreStats() []byte { return s.v2store.JsonStats() }

func (s *EtcdServer) checkMembershipOperationPermission(ctx context.Context) error {
	if s.authStore == nil {
		// 在普通etcd进程的上下文中，s.authStore永远不会为零。
		// 此分支用于处理服务器测试中的案例。go 
		return nil
	}

	// 请注意，此权限检查是在API层中完成的，
	// 所以TOCTOU问题可能会在这样的计划中引起：
	// 使用用户a更新成员资格->撤销a的根角色->应用成员资格更改
	// 在状态机层
	// 但是，成员身份更改和角色管理都需要根权限。
	// 所以管理员的谨慎操作可以防止问题的发生。
	authInfo, err := s.AuthInfoFromCtx(ctx)
	if err != nil {
		return err
	}

	return s.AuthStore().IsAdminPermitted(authInfo)
}

func (s *EtcdServer) AddMember(ctx context.Context, memb membership.Member) ([]*membership.Member, error) {
	if err := s.checkMembershipOperationPermission(ctx); err != nil {
		return nil, err
	}

	// TODO:将成员移动到protobuf类型
	b, err := json.Marshal(memb)
	if err != nil {
		return nil, err
	}

	// 默认启用StrictReconfigCheck；如果不健康，拒绝新成员。
	if err := s.mayAddMember(memb); err != nil {
		return nil, err
	}

	cc := raftpb.ConfChange{
		Type:    raftpb.ConfChangeAddNode,
		NodeID:  uint64(memb.ID),
		Context: b,
	}

	if memb.IsLearner {
		cc.Type = raftpb.ConfChangeAddLearnerNode
	}

	return s.configure(ctx, cc)
}

func (s *EtcdServer) mayAddMember(memb membership.Member) error {
	lg := s.Logger()
	if !s.Cfg.StrictReconfigCheck {
		return nil
	}

	// 添加投票成员时保护法定人数
	if !memb.IsLearner && !s.cluster.IsReadyToAddVotingMember() {
		lg.Warn(
			"rejecting member add request; not enough healthy members",
			zap.String("local-member-id", s.ID().String()),
			zap.String("requested-member-add", fmt.Sprintf("%+v", memb)),
			zap.Error(ErrNotEnoughStartedMembers),
		)
		return ErrNotEnoughStartedMembers
	}

	if !isConnectedFullySince(s.r.transport, time.Now().Add(-HealthInterval), s.ID(), s.cluster.VotingMembers()) {
		lg.Warn(
			"rejecting member add request; local member has not been connected to all peers, reconfigure breaks active quorum",
			zap.String("local-member-id", s.ID().String()),
			zap.String("requested-member-add", fmt.Sprintf("%+v", memb)),
			zap.Error(ErrUnhealthy),
		)
		return ErrUnhealthy
	}

	return nil
}

func (s *EtcdServer) RemoveMember(ctx context.Context, id uint64) ([]*membership.Member, error) {
	if err := s.checkMembershipOperationPermission(ctx); err != nil {
		return nil, err
	}

	// 默认启用StrictReconfigCheck；如果导致法定人数丢失，则拒绝删除
	if err := s.mayRemoveMember(types.ID(id)); err != nil {
		return nil, err
	}

	cc := raftpb.ConfChange{
		Type:   raftpb.ConfChangeRemoveNode,
		NodeID: id,
	}
	return s.configure(ctx, cc)
}

// PromoteMember将学习者节点升级为投票节点。
func (s *EtcdServer) PromoteMember(ctx context.Context, id uint64) ([]*membership.Member, error) {
	// 只有筏长有待提升学习者节点是否准备就绪的信息。如果promoteMember调用
	// 失败，错误为ErrNotLeader，则通过HTTP将请求转发给leader节点。如果promoteMember调用失败并出现错误
	// 而不是ErrNotLeader，则返回错误。
	resp, err := s.promoteMember(ctx, id)
	if err == nil {
		learnerPromoteSucceed.Inc()
		return resp, nil
	}
	if err != ErrNotLeader {
		learnerPromoteFailed.WithLabelValues(err.Error()).Inc()
		return resp, err
	}

	cctx, cancel := context.WithTimeout(ctx, s.Cfg.ReqTimeout())
	defer cancel()
	// 转发给领导
	for cctx.Err() == nil {
		leader, err := s.waitLeader(cctx)
		if err != nil {
			return nil, err
		}
		for _, url := range leader.PeerURLs {
			resp, err := promoteMemberHTTP(cctx, url, id, s.peerRt)
			if err == nil {
				return resp, nil
			}
			// 如果会员升级失败，请提前返回。否则请继续重试。
			if err == ErrLearnerNotReady || err == membership.ErrIDNotFound || err == membership.ErrMemberNotLearner {
				return nil, err
			}
		}
	}

	if cctx.Err() == context.DeadlineExceeded {
		return nil, ErrTimeout
	}
	return nil, ErrCanceled
}

// promoteMember在向raft发送promote 
// 请求之前，检查要升级的学习者节点是否准备就绪。
// 如果本地节点不是raft leader（因此没有足够的信息来确定学习者节点是否准备就绪），则函数返回ErrNotLeader；如果
// 本地节点是leader（因此有足够的信息），则函数返回ErrLearnerNotReady但决定学习者节点尚未准备好升级
// 。
func (s *EtcdServer) promoteMember(ctx context.Context, id uint64) ([]*membership.Member, error) {
	if err := s.checkMembershipOperationPermission(ctx); err != nil {
		return nil, err
	}

	// 检查我们是否可以提升该学员。
	if err := s.mayPromoteMember(types.ID(id)); err != nil {
		return nil, err
	}

	// 为更改创建上下文。马克·伊斯莱纳为假，我为真。
	promoteChangeContext := membership.ConfigChangeContext{
		Member: membership.Member{
			ID: types.ID(id),
		},
		IsPromote: true,
	}

	b, err := json.Marshal(promoteChangeContext)
	if err != nil {
		return nil, err
	}

	cc := raftpb.ConfChange{
		Type:    raftpb.ConfChangeAddNode,
		NodeID:  id,
		Context: b,
	}

	return s.configure(ctx, cc)
}

func (s *EtcdServer) mayPromoteMember(id types.ID) error {
	lg := s.Logger()
	err := s.isLearnerReady(uint64(id))
	if err != nil {
		return err
	}

	if !s.Cfg.StrictReconfigCheck {
		return nil
	}
	if !s.cluster.IsReadyToPromoteMember(uint64(id)) {
		lg.Warn(
			"rejecting member promote request; not enough healthy members",
			zap.String("local-member-id", s.ID().String()),
			zap.String("requested-member-remove-id", id.String()),
			zap.Error(ErrNotEnoughStartedMembers),
		)
		return ErrNotEnoughStartedMembers
	}

	return nil
}

// 检查学习者是否赶上领导。
// 注意：如果在集群中找不到成员或成员不是学习者，则返回nil。
// 这两个条件将在稍后的应用阶段之前进行检查。
func (s *EtcdServer) isLearnerReady(id uint64) error {
	rs := s.raftStatus()

	// 领导者的身份。进步不是零
	if rs.Progress == nil {
		return ErrNotLeader
	}

	var learnerMatch uint64
	isFound := false
	leaderID := rs.ID
	for memberID, progress := range rs.Progress {
		if id == memberID {
			// 检查其状态
			learnerMatch = progress.Match
			isFound = true
			break
		}
	}

	if isFound {
		leaderMatch := rs.Progress[leaderID].Match
		// 学习者的比赛尚未赶上领先者
		if float64(learnerMatch) < float64(leaderMatch)*readyPercent {
			return ErrLearnerNotReady
		}
	}

	return nil
}

func (s *EtcdServer) mayRemoveMember(id types.ID) error {
	if !s.Cfg.StrictReconfigCheck {
		return nil
	}

	lg := s.Logger()
	isLearner := s.cluster.IsMemberExist(id) && s.cluster.Member(id).IsLearner
	// 删除无表决权成员
	if isLearner {
		return nil
	}

	if !s.cluster.IsReadyToRemoveVotingMember(uint64(id)) {
		lg.Warn(
			"rejecting member remove request; not enough healthy members",
			zap.String("local-member-id", s.ID().String()),
			zap.String("requested-member-remove-id", id.String()),
			zap.Error(ErrNotEnoughStartedMembers),
		)
		return ErrNotEnoughStartedMembers
	}

	// 删除被删除的成员是安全的，因为它不属于活动的法定人数
	if t := s.r.transport.ActiveSince(id); id != s.ID() && t.IsZero() {
		return nil
	}

	// 如果一些成员关闭了
	m := s.cluster.VotingMembers()
	active := numConnectedSince(s.r.transport, time.Now().Add(-HealthInterval), s.ID(), m)
	if (active - 1) < 1+((len(m)-1)/2) {
		lg.Warn(
			"rejecting member remove request; local member has not been connected to all peers, reconfigure breaks active quorum",
			zap.String("local-member-id", s.ID().String()),
			zap.String("requested-member-remove", id.String()),
			zap.Int("active-peers", active),
			zap.Error(ErrUnhealthy),
		)
		return ErrUnhealthy
	}

	return nil
}

func (s *EtcdServer) UpdateMember(ctx context.Context, memb membership.Member) ([]*membership.Member, error) {
	b, merr := json.Marshal(memb)
	if merr != nil {
		return nil, merr
	}

	if err := s.checkMembershipOperationPermission(ctx); err != nil {
		return nil, err
	}
	cc := raftpb.ConfChange{
		Type:    raftpb.ConfChangeUpdateNode,
		NodeID:  uint64(memb.ID),
		Context: b,
	}
	return s.configure(ctx, cc)
}

func (s *EtcdServer) setCommittedIndex(v uint64) {
	atomic.StoreUint64(&s.committedIndex, v)
}

func (s *EtcdServer) getCommittedIndex() uint64 {
	return atomic.LoadUint64(&s.committedIndex)
}

func (s *EtcdServer) setAppliedIndex(v uint64) {
	atomic.StoreUint64(&s.appliedIndex, v)
}

func (s *EtcdServer) getAppliedIndex() uint64 {
	return atomic.LoadUint64(&s.appliedIndex)
}

func (s *EtcdServer) setTerm(v uint64) {
	atomic.StoreUint64(&s.term, v)
}

func (s *EtcdServer) getTerm() uint64 {
	return atomic.LoadUint64(&s.term)
}

func (s *EtcdServer) setLead(v uint64) {
	atomic.StoreUint64(&s.lead, v)
}

func (s *EtcdServer) getLead() uint64 {
	return atomic.LoadUint64(&s.lead)
}

func (s *EtcdServer) LeaderChangedNotify() <-chan struct{} {
	return s.leaderChanged.Receive()
}

// FIRSTCOMITENTERMTNOTIFY返回频道，该频道将在第一个
// 新学期提交的条目上解锁，这是新领导回答
// 只读请求所必需的（leader无法响应任何只读请求
// 只要需要线性语义）
func (s *EtcdServer) FirstCommitInTermNotify() <-chan struct{} {
	return s.firstCommitInTerm.Receive()
}

// RaftStatusGetter表示etcd服务器和Raft进度。
type RaftStatusGetter interface {
	ID() types.ID
	Leader() types.ID
	CommittedIndex() uint64
	AppliedIndex() uint64
	Term() uint64
}

func (s *EtcdServer) ID() types.ID { return s.id }

func (s *EtcdServer) Leader() types.ID { return types.ID(s.getLead()) }

func (s *EtcdServer) Lead() uint64 { return s.getLead() }

func (s *EtcdServer) CommittedIndex() uint64 { return s.getCommittedIndex() }

func (s *EtcdServer) AppliedIndex() uint64 { return s.getAppliedIndex() }

func (s *EtcdServer) Term() uint64 { return s.getTerm() }

type confChangeResponse struct {
	membs []*membership.Member
	err   error
}

// configure通过协商一致发送配置更改，然后
// 等待将其应用于服务器。它将阻塞，直到执行更改或出现错误。
func (s *EtcdServer) configure(ctx context.Context, cc raftpb.ConfChange) ([]*membership.Member, error) {
	lg := s.Logger()
	cc.ID = s.reqIDGen.Next()
	ch := s.w.Register(cc.ID)

	start := time.Now()
	if err := s.r.ProposeConfChange(ctx, cc); err != nil {
		s.w.Trigger(cc.ID, nil)
		return nil, err
	}

	select {
	case x := <-ch:
		if x == nil {
			lg.Panic("failed to configure")
		}
		resp := x.(*confChangeResponse)
		lg.Info(
			"applied a configuration change through raft",
			zap.String("local-member-id", s.ID().String()),
			zap.String("raft-conf-change", cc.Type.String()),
			zap.String("raft-conf-change-node-id", types.ID(cc.NodeID).String()),
		)
		return resp.membs, resp.err

	case <-ctx.Done():
		s.w.Trigger(cc.ID, nil) // GC wait
		return nil, s.parseProposeCtxErr(ctx.Err(), start)

	case <-s.stopping:
		return nil, ErrStopped
	}
}

// 在 run goroutine 中 会通过 getSyncC（） 函数监听该定时器 ， 当定 时器 到 期时会调用
// EtcdServer.sync（） 方法发送 SYNC 消息
func (s *EtcdServer) sync(timeout time.Duration) {
	req := pb.Request{	// 创建 SYNC 消息
		Method: "SYNC",
		ID:     s.reqIDGen.Next(),
		Time:   time.Now().UnixNano(),
	}
	data := pbutil.MustMarshal(&req)	// 序列化 SYNC 消息
	// 不保证在执行同步请求时节点有前导，
	// so它使用goroutine进行建议。
	ctx, cancel := context.WithTimeout(s.ctx, timeout)
	s.GoAttach(func() {	// 启动一个后台 goroutine ，发送 SYNC (MsgProp ）消息
		s.r.Propose(ctx, data)
		cancel()
	})
}

// publishV3使用v3请求将服务器信息注册到集群中。
// 信息是此服务器成员结构的JSON表示形式，使用服务器的静态ClientURL更新了
// 函数不断尝试注册成功之前，
// 或其服务器停止。
func (s *EtcdServer) publishV3(timeout time.Duration) {
	req := &membershippb.ClusterMemberAttrSetRequest{
		Member_ID: uint64(s.id),
		MemberAttributes: &membershippb.Attributes{
			Name:       s.attributes.Name,
			ClientUrls: s.attributes.ClientURLs,
		},
	}
	lg := s.Logger()
	for {
		select {
		case <-s.stopping:
			lg.Warn(
				"stopped publish because server is stopping",
				zap.String("local-member-id", s.ID().String()),
				zap.String("local-member-attributes", fmt.Sprintf("%+v", s.attributes)),
				zap.Duration("publish-timeout", timeout),
			)
			return

		default:
		}

		ctx, cancel := context.WithTimeout(s.ctx, timeout)
		_, err := s.raftRequest(ctx, pb.InternalRaftRequest{ClusterMemberAttrSet: req})
		cancel()
		switch err {
		case nil:
			close(s.readych)
			lg.Info(
				"published local member to cluster through raft",
				zap.String("local-member-id", s.ID().String()),
				zap.String("local-member-attributes", fmt.Sprintf("%+v", s.attributes)),
				zap.String("cluster-id", s.cluster.ID().String()),
				zap.Duration("publish-timeout", timeout),
			)
			return

		default:
			lg.Warn(
				"failed to publish local member to cluster through raft",
				zap.String("local-member-id", s.ID().String()),
				zap.String("local-member-attributes", fmt.Sprintf("%+v", s.attributes)),
				zap.Duration("publish-timeout", timeout),
				zap.Error(err),
			)
		}
	}
}

// 发布将服务器信息注册到集群中。信息
// 是此服务器成员结构的JSON表示，使用服务器的
// 静态客户端URL更新。
// 函数n一直尝试注册，直到注册成功，
// 或其服务器停止。
// 
// 使用v2存储编码成员属性，并通过Raft 
// 应用，但不通过v2 API端点，这意味着即使禁用了v2 
// 客户端处理程序（例如--enable-v2=false），集群仍然可以
// 通过RAFT处理发布请求http
//
// EtcdServer.Start（） 方法中会启动一个后台 goroutine 将当前节点的相关信息发送到集群其
// 他节点 （ 即将当前节点注册到集群当中），相应的逻辑位于 EtcdServer. publish（）方法中
// TODO:在3.6中删除（开始使用publishV3）
func (s *EtcdServer) publish(timeout time.Duration) {
	lg := s.Logger()
	// 将 EtcdServer.attributes 字段序列化成 JSON 格式，
	b, err := json.Marshal(s.attributes)
	if err != nil {
		lg.Panic("failed to marshal JSON", zap.Error(err))
		return
	}
	req := pb.Request{	// 将上述 JSON 数据封装成 PUT 请求
		Method: "PUT",
		Path:   membership.MemberAttributesStorePath(s.id),
		Val:    string(b),
	}

	for {
		ctx, cancel := context.WithTimeout(s.ctx, timeout)
		_, err := s.Do(ctx, req)	// 调用 EtcdServer.DO （）方法处理该请求
		cancel()
		switch err {
		case nil:
			// 将当前节点信息发送到集群其他节点之后，会将 readych 远远关闭，从而实现通知其他 goroutine 的目的，
			close(s.readych)
			lg.Info(
				"published local member to cluster through raft",
				zap.String("local-member-id", s.ID().String()),
				zap.String("local-member-attributes", fmt.Sprintf("%+v", s.attributes)),
				zap.String("request-path", req.Path),
				zap.String("cluster-id", s.cluster.ID().String()),
				zap.Duration("publish-timeout", timeout),
			)
			return
			// 如果出现错误，则会输出相应的日志，然后继续当前的 for 循环 ， 直至注册成功

		case ErrStopped:
			lg.Warn(
				"stopped publish because server is stopped",
				zap.String("local-member-id", s.ID().String()),
				zap.String("local-member-attributes", fmt.Sprintf("%+v", s.attributes)),
				zap.Duration("publish-timeout", timeout),
				zap.Error(err),
			)
			return

		default:
			lg.Warn(
				"failed to publish local member to cluster through raft",
				zap.String("local-member-id", s.ID().String()),
				zap.String("local-member-attributes", fmt.Sprintf("%+v", s.attributes)),
				zap.String("request-path", req.Path),
				zap.Duration("publish-timeout", timeout),
				zap.Error(err),
			)
		}
	}
}

// 创建完 snap . Message 实例之后会调用 EtcdServer.sendMergedSnap（） 方法将其发送到指定节点
func (s *EtcdServer) sendMergedSnap(merged snap.Message) {
	// 递增 inflightSnapshots 字段， 它表示已发送但未收到响应的快照消息个数
	atomic.AddInt64(&s.inflightSnapshots, 1)

	lg := s.Logger()
	fields := []zap.Field{
		zap.String("from", s.ID().String()),
		zap.String("to", types.ID(merged.To).String()),
		zap.Int64("bytes", merged.TotalSize),
		zap.String("size", humanize.Bytes(uint64(merged.TotalSize))),
	}

	now := time.Now()
	// 发送 snap.Message 消息 ， 底层会启动羊独的后台 goroutine ， 通过 snapshotSender 完成发送 ，
	s.r.transport.SendSnapshot(merged)
	lg.Info("sending merged snapshot", fields...)

	s.GoAttach(func() {	// 启动一个后台 goroutine 监听该快照消息是否发送完成
		select {
		case ok := <-merged.CloseNotify():
			// 延迟发布机上快照30秒，以
			// 阻止日志压缩。
			// 如果跟随者仍然赶不上，则可能赶不上。无论如何，我们无法避免快照周期。
			if ok {
				select {
				case <-time.After(releaseDelayAfterSnapshot):	// 默认阻塞等待 30 秒
				case <-s.stopping:
				}
			}
			// snap.Message 消息发送完成（或是超时）之后会递减 inflightSnapshots 佳，
			// 当 inflightSnapshots 递减到 0 时 ， 前面对 MernoryStorage 的压缩才能执行
			atomic.AddInt64(&s.inflightSnapshots, -1)

			lg.Info("sent merged snapshot", append(fields, zap.Duration("took", time.Since(now)))...)

		case <-s.stopping:
			lg.Warn("canceled sending merged snapshot; server stopping", fields...)
			return
		}
	})
}

// 在 apply() 方法中会遍历 ents 中的全部 Entry 记录，并根据 Entry 的类型进行不同的处理。
func (s *EtcdServer) apply(
	es []raftpb.Entry,
	confState *raftpb.ConfState,
) (appliedt uint64, appliedi uint64, shouldStop bool) {
	s.lg.Debug("Applying entries", zap.Int("num-entries", len(es)))
	for i := range es {	// 遍历待应用的 Entry 记录
		e := es[i]
		s.lg.Debug("Applying entry",
			zap.Uint64("index", e.Index),
			zap.Uint64("term", e.Term),
			zap.Stringer("type", e.Type))
		switch e.Type {	// 根据 Entry 记录的 不同类型，进行不同的处理
		case raftpb.EntryNormal:
			s.applyEntryNormal(&e)
			s.setAppliedIndex(e.Index)
			s.setTerm(e.Term)

		case raftpb.EntryConfChange:
			// 我们需要在v2store 
			// 之上应用所有WAL条目，并且只应用“未应用”（例如Index>backend.ConsistentIndex）在后端.
			shouldApplyV3 := membership.ApplyV2storeOnly

			// 更新 EtcdServer.consistindex ，其中保存了当前节点应用的最后一条 Entry 记录的索引值
			if e.Index > s.consistIndex.ConsistentIndex() {
				s.consistIndex.SetConsistentIndex(e.Index, e.Term)
				shouldApplyV3 = membership.ApplyBoth
			}

			var cc raftpb.ConfChange
			pbutil.MustUnmarshal(&cc, e.Data)	// 将 Entry.Data 反序列化成 ConfChange 实例
			// 调用 applyConfChange （）方法处理 ConfChange ，注意返回值 removedSelf ，当它为 true
			// 时表示将当前节点从集群中移除
			removedSelf, err := s.applyConfChange(cc, confState, shouldApplyV3)
			s.setAppliedIndex(e.Index)	// 更新 EtcdServer.appliedindex 字段
			s.setTerm(e.Term)
			shouldStop = shouldStop || removedSelf
			// 它们都会阻塞监听 ConfChange 对应的通道，这里在对应的 Entry 处理完成之后，
			// 会关闭对应的通道， 通知监听的 goroutine
			s.w.Trigger(cc.ID, &confChangeResponse{s.cluster.Members(), err})

		default:
			lg := s.Logger()
			lg.Panic(
				"unknown entry type; must be either EntryNormal or EntryConfChange",
				zap.String("type", e.Type.String()),
			)
		}
		appliedi, appliedt = e.Index, e.Term	//更新 appliedt 、 appliedi 返回值
	}
	return appliedt, appliedi, shouldStop
}

// applyEntryNormal将EntryNormal类型raftpb请求应用于EtcdServer
// EtcdServer.applyEntryNormal（） 方法处理 EntryNormal 记录的具体过程。
// applyEntryNormal （） 方法首先会尝试将 Entry.Data 反序列化成 IntemalRaftRequest 实例 ， 如果失败，
// 则将其反序列化成 etcdserverpb. Request 实例，之后根据反序列化的结果调用 EtcdServer 的相应
// 方法进行处理， 最后将处理结果写入 Entry 对应的通道中 。
func (s *EtcdServer) applyEntryNormal(e *raftpb.Entry) {
	shouldApplyV3 := membership.ApplyV2storeOnly
	index := s.consistIndex.ConsistentIndex()
	if e.Index > index {
		// 设置当前执行项的一致索引
		s.consistIndex.SetConsistentIndex(e.Index, e.Term)	// 更新 EtcdServer.consistindex 记录的索引值

		shouldApplyV3 = membership.ApplyBoth
	}
	s.lg.Debug("apply entry normal",
		zap.Uint64("consistent-index", index),
		zap.Uint64("entry-index", e.Index),
		zap.Bool("should-applyV3", bool(shouldApplyV3)))

	if len(e.Data) == 0 {	// 空的 Entry 记录只会在 Leader 选举结束时 出 现
		s.firstCommitInTerm.Notify()

		if s.isLeader() {	// 如果当前节点为 Leader ，则晋升其 lessor 实例
			s.lessor.Promote(s.Cfg.ElectionTimeout())
		}
		return
	}

	var raftReq pb.InternalRaftRequest
	// 尝试将 Entry.Data 反序列化成 InternalRaftRequest 实例 ， InternalRaftRequest 中封装了所
	// 有类型的 Client 请求
	if !pbutil.MaybeUnmarshal(&raftReq, e.Data) { // 向后兼容
		var r pb.Request
		rp := &r
		// 兼容性处理，如果上述序列化失败，则将 Entry.Date 反序列化成 pb.Request 实例
		pbutil.MustUnmarshal(rp, e.Data)
		s.lg.Debug("applyEntryNormal", zap.Stringer("V2request", rp))
		// 调用 EtcdServer.applyV2Request （）方法进行处理，在 applyV2Request （） 方法中，
		// 会根据请求的类型，调用不同的方法进行处理。处理结束后会将结采写入 Wait 中记录对应的通道中，
		// 然后关闭该通道，
		s.w.Trigger(r.ID, s.applyV2Request((*RequestV2)(rp), shouldApplyV3))
		return
	}
	s.lg.Debug("applyEntryNormal", zap.Stringer("raftReq", &raftReq))

	if raftReq.V2 != nil {	// 上述序列化成功，且是 v2 版本的请求，调用 applyV2Request （）方法处理
		req := (*RequestV2)(raftReq.V2)
		// 关闭 Wait 中记录的该 Entry 对应的通道
		s.w.Trigger(req.ID, s.applyV2Request(req, shouldApplyV3))
		return
	}
	// 下面是对 v3 版本请求的处理
	id := raftReq.ID	// 获取请求 id
	if id == 0 {
		id = raftReq.Header.ID
	}

	var ar *applyResult
	needResult := s.w.IsRegistered(id)	// 检测该请求是否需要进行响应
	if needResult || !noSideEffect(&raftReq) {
		if !needResult && raftReq.Txn != nil {
			removeNeedlessRangeReqs(raftReq.Txn)
		}
		// 调用 applyV3.Apply （）方法处理该 Entry ，其中会根据请求的类型选择不同的方法进行处理
		ar = s.applyV3.Apply(&raftReq, shouldApplyV3)
	}

	// 不要重新应用应用的条目。
	if !shouldApplyV3 {
		return
	}
	// 返回结果 ar(applyResult 类型）为nil ，直接返回（
	if ar == nil {
		return
	}
	// 如果返回了 ErrNoSpace 错误，则表示底层的 Backend 已经没有足够的空间，如果是第 一次出现这种情
	// 况 ， 则在后面立即启动一个后台 goroutine ，并调用 EtcdServer.raftRequest （）方法发送
	// AlarmRequest 请求，当前其他节点收到该请求时 ， 会停止后续的 PUT 操作
	if ar.err != ErrNoSpace || len(s.alarmStore.Get(pb.AlarmType_NOSPACE)) > 0 {
		s.w.Trigger(id, ar)	// 将上述处理结果写入对应的通道中 ， 然后将对应通道关 闭
		return
	}

	lg := s.Logger()
	lg.Warn(
		"message exceeded backend quota; raising alarm",
		zap.Int64("quota-size-bytes", s.Cfg.QuotaBackendBytes),
		zap.String("quota-size", humanize.Bytes(uint64(s.Cfg.QuotaBackendBytes))),
		zap.Error(ar.err),
	)

	s.GoAttach(func() {	// 第一次出现 ErrNoSpace 错误
		a := &pb.AlarmRequest{	// 创建 AlarmRequest
			MemberID: uint64(s.ID()),
			Action:   pb.AlarmRequest_ACTIVATE,
			Alarm:    pb.AlarmType_NOSPACE,
		}
		// 将 AlarmRequest 请求封装成 MsgProp 消息，并友送到集群中
		s.raftRequest(s.ctx, pb.InternalRaftRequest{Alarm: a})
		s.w.Trigger(id, ar)	// 将上述处理结果写入对应的通道中，然后将对应通过关闭
	})
}

// EntryConfChange 类型的 Entry 记录主要是通过 EtcdServer.applyConfChange （） 方法进行处理
//的，在 applyConfChange（）方法中会根据 ConfChange 的类型进行分类处理
func (s *EtcdServer) applyConfChange(cc raftpb.ConfChange, confState *raftpb.ConfState, shouldApplyV3 membership.ShouldApplyV3) (bool, error) {
	// 在开始进行节点修改之前，先调用 ValidateConfigurationChange （）方法进行检测，
	if err := s.cluster.ValidateConfigurationChange(cc); err != nil {
		cc.NodeID = raft.None
		s.r.ApplyConfChange(cc)
		return false, err
	}

	lg := s.Logger()
	// 将 ConfChange 交给 etcd-raft 模块进行处理，其中会根据 ConfChange 的类型进行分类处理，
	// 这里的返回值的 ConfState 中记录了集群中最新的节点 id
	*confState = *s.r.ApplyConfChange(cc)
	s.beHooks.SetConfState(confState)
	switch cc.Type {	// 根据请求的类型进行相应的处理
	case raftpb.ConfChangeAddNode, raftpb.ConfChangeAddLearnerNode:
		// 将 ConfChange.Context 中的数据反序列 化成 Member 实例 ，
		confChangeContext := new(membership.ConfigChangeContext)
		if err := json.Unmarshal(cc.Context, confChangeContext); err != nil {
			lg.Panic("failed to unmarshal member", zap.Error(err))
		}
		if cc.NodeID != uint64(confChangeContext.Member.ID) {
			lg.Panic(
				"got different member ID",
				zap.String("member-id-from-config-change-entry", types.ID(cc.NodeID).String()),
				zap.String("member-id-from-message", confChangeContext.Member.ID.String()),
			)
		}
		if confChangeContext.IsPromote {
			s.cluster.PromoteMember(confChangeContext.Member.ID, shouldApplyV3)
		} else {
			// 将新 Member 实例添加到本地 RaftCluster 中
			s.cluster.AddMember(&confChangeContext.Member, shouldApplyV3)

			if confChangeContext.Member.ID != s.id {
				// 如果添加的是远端节点，则需妥在 Transport 中添加对应的 Peer 实例 ，
				s.r.transport.AddPeer(confChangeContext.Member.ID, confChangeContext.PeerURLs)
			}
		}

		// 的ConfChange调用的。当此服务器id等于Raft成员ConfChange 
		if confChangeContext.Member.ID == s.id {
			if cc.Type == raftpb.ConfChangeAddLearnerNode {
				isLearner.Set(1)
			} else {
				isLearner.Set(0)
			}
		}

	case raftpb.ConfChangeRemoveNode:
		id := types.ID(cc.NodeID)
		s.cluster.RemoveMember(id, shouldApplyV3)	//  从 RaftCluster 中删除指定的 Member 实例
		if id == s.id {
			return true, nil	// 如果移除的是当前节点则返回 true
		}
		s.r.transport.RemovePeer(id)	// 若移除远梢节点，则需妥从 Transporter 中删除对应的 Peer 实例

	case raftpb.ConfChangeUpdateNode:
		// 将 ConfChange.Context 中的数据反序列化成 Member 实例，
		m := new(membership.Member)
		if err := json.Unmarshal(cc.Context, m); err != nil {
			lg.Panic("failed to unmarshal member", zap.Error(err))
		}
		if cc.NodeID != uint64(m.ID) {
			lg.Panic(
				"got different member ID",
				zap.String("member-id-from-config-change-entry", types.ID(cc.NodeID).String()),
				zap.String("member-id-from-message", m.ID.String()),
			)
		}
		// 更新本地 RaftCluster 实例中相应的 Member 实例
		s.cluster.UpdateRaftAttributes(m.ID, m.RaftAttributes, shouldApplyV3)
		if m.ID != s.id {	// 如果是更新的是远端节点 ，则需要更新 Transporter 中对应的 Peer 实例
			s.r.transport.UpdatePeer(m.ID, m.PeerURLs)
		}
	}
	return false, nil
}

// EtcdServer.snapshot （） 方法是真正生成快照文件的地方，其中会启动一个单独的后台
// goroutine 来完成新快照文件的生成，主要是序列化 v2 存储中的数据并持久化到文件中，触发相
// 应的压缩操作
func (s *EtcdServer) snapshot(snapi uint64, confState raftpb.ConfState) {
	clone := s.v2store.Clone()	// 复制 v2 存储
	s.KV().Commit()	// 提交 v3 存储中当前等待读写事务

	s.GoAttach(func() {
		lg := s.Logger()

		d, err := clone.SaveNoCopy()	// 将 v2 存储序列化成 JSON 数据
		if err != nil {
			lg.Panic("failed to save v2 store", zap.Error(err))
		}
		// 将上述快照数据和元数据更新到 etcd-raft 模块中的 MemoryStorage 中，
		// 并且返回 Snapshot 实例（即 MemoryStorage.snapshot 字段）
		snap, err := s.r.raftStorage.CreateSnapshot(snapi, &confState, d)
		if err != nil {
			// 快照是与raft的进度异步完成的。
			// raft可能已经有了更新的快照。
			if err == raft.ErrSnapOutOfDate {
				return
			}
			lg.Panic("failed to create snapshot", zap.Error(err))
		}
		// 将 v2 存储的快照数据记录到磁盘中，该过程涉及在 WAL 日志文件中记录快照元数据及写入 snap 文件等操作 ，
		if err = s.r.storage.SaveSnap(snap); err != nil {
			lg.Panic("failed to save snapshot", zap.Error(err))
		}
		if err = s.r.storage.Release(snap); err != nil {
			lg.Panic("failed to release wal", zap.Error(err))
		}

		lg.Info(
			"saved snapshot",
			zap.Uint64("snapshot-index", snap.Metadata.Index),
		)

		// 发送快照时，etcd将暂停压缩。
		// 在接收到快照后，慢跟随者需要在发送快照后立即获取所有条目，以赶上进度。如果不暂停压缩，则在
		// 发送的快照之后的日志项可能已经被压缩。当快照需要很长时间
		// 才能发送和保存时，就会发生这种情况。暂停压缩可避免触发快照发送周期。
		//
		// 如果当前还存在已发送但未响应的快照消息，则不能进行后续的压缩操作 ， 如果进行了后续的压缩，
		// 则可能导致 Follower 节点再次无法追赶上 Leader 节点，从而需妥再次发送快照数据
		if atomic.LoadInt64(&s.inflightSnapshots) != 0 {
			lg.Info("skip compaction since there is an inflight snapshot")
			return
		}

		// 为了防止集群中存在比较慢的 Follower 节点，保留 5000 条 Entry 记录不压缩
		compacti := uint64(1)
		if snapi > s.Cfg.SnapshotCatchUpEntries {
			compacti = snapi - s.Cfg.SnapshotCatchUpEntries
		}
		// 压缩 MemoryStorage 中的指定位置之前的全部 Entry 记录 ， 不再展开
		err = s.r.raftStorage.Compact(compacti)
		if err != nil {
			// 压实与筏板的进度同步进行。
			// 木筏日志可能已经很紧凑。
			if err == raft.ErrCompacted {
				return
			}
			lg.Panic("failed to compact", zap.Error(err))
		}
		lg.Info(
			"compacted Raft logs",
			zap.Uint64("compact-index", compacti),
		)
	})
}

// CutPeer将消息丢弃到指定的对等方。
func (s *EtcdServer) CutPeer(id types.ID) {
	tr, ok := s.r.transport.(*rafthttp.Transport)
	if ok {
		tr.CutPeer(id)
	}
}

// MendPeer恢复给定对等方的消息丢弃行为。
func (s *EtcdServer) MendPeer(id types.ID) {
	tr, ok := s.r.transport.(*rafthttp.Transport)
	if ok {
		tr.MendPeer(id)
	}
}

func (s *EtcdServer) PauseSending() { s.r.pauseSending() }

func (s *EtcdServer) ResumeSending() { s.r.resumeSending() }

func (s *EtcdServer) ClusterVersion() *semver.Version {
	if s.cluster == nil {
		return nil
	}
	return s.cluster.Version()
}

// monitorClusterVersions每个monitorVersionInterval都会检查它是否是领导者，并在需要时更新群集版本。
func (s *EtcdServer) monitorClusterVersions() {
	monitor := serverversion.NewMonitor(s.Logger(), newServerVersionAdapter(s))
	for {
		select {
		case <-s.firstCommitInTerm.Receive():
		case <-time.After(monitorVersionInterval):
		case <-s.stopping:
			return
		}

		if s.Leader() != s.ID() {
			continue
		}
		monitor.UpdateClusterVersionIfNeeded()
	}
}

// monitorStorageVersion如果需要，每个monitorVersionInterval都会更新存储版本。
func (s *EtcdServer) monitorStorageVersion() {
	monitor := serverversion.NewMonitor(s.Logger(), newServerVersionAdapter(s))
	for {
		select {
		case <-time.After(monitorVersionInterval):
		case <-s.clusterVersionChanged.Receive():
		case <-s.stopping:
			return
		}
		monitor.UpdateStorageVersionIfNeeded()
	}
}

func (s *EtcdServer) updateClusterVersionV2(ver string) {
	lg := s.Logger()

	if s.cluster.Version() == nil {
		lg.Info(
			"setting up initial cluster version using v2 API",
			zap.String("cluster-version", version.Cluster(ver)),
		)
	} else {
		lg.Info(
			"updating cluster version using v2 API",
			zap.String("from", version.Cluster(s.cluster.Version().String())),
			zap.String("to", version.Cluster(ver)),
		)
	}

	req := pb.Request{
		Method: "PUT",
		Path:   membership.StoreClusterVersionKey(),
		Val:    ver,
	}

	ctx, cancel := context.WithTimeout(s.ctx, s.Cfg.ReqTimeout())
	_, err := s.Do(ctx, req)
	cancel()

	switch err {
	case nil:
		lg.Info("cluster version is updated", zap.String("cluster-version", version.Cluster(ver)))
		return

	case ErrStopped:
		lg.Warn("aborting cluster version update; server is stopped", zap.Error(err))
		return

	default:
		lg.Warn("failed to update cluster version", zap.Error(err))
	}
}

func (s *EtcdServer) updateClusterVersionV3(ver string) {
	lg := s.Logger()

	if s.cluster.Version() == nil {
		lg.Info(
			"setting up initial cluster version using v3 API",
			zap.String("cluster-version", version.Cluster(ver)),
		)
	} else {
		lg.Info(
			"updating cluster version using v3 API",
			zap.String("from", version.Cluster(s.cluster.Version().String())),
			zap.String("to", version.Cluster(ver)),
		)
	}

	req := membershippb.ClusterVersionSetRequest{Ver: ver}

	ctx, cancel := context.WithTimeout(s.ctx, s.Cfg.ReqTimeout())
	_, err := s.raftRequest(ctx, pb.InternalRaftRequest{ClusterVersionSet: &req})
	cancel()

	switch err {
	case nil:
		lg.Info("cluster version is updated", zap.String("cluster-version", version.Cluster(ver)))
		return

	case ErrStopped:
		lg.Warn("aborting cluster version update; server is stopped", zap.Error(err))
		return

	default:
		lg.Warn("failed to update cluster version", zap.Error(err))
	}
}

// 监视器降级每次降级检查时间检查是否是领导者，如果需要，取消降级。
func (s *EtcdServer) monitorDowngrade() {
	monitor := serverversion.NewMonitor(s.Logger(), newServerVersionAdapter(s))
	t := s.Cfg.DowngradeCheckTime
	if t == 0 {
		return
	}
	for {
		select {
		case <-time.After(t):
		case <-s.stopping:
			return
		}

		if !s.isLeader() {
			continue
		}
		monitor.CancelDowngradeIfNeeded()
	}
}

func (s *EtcdServer) parseProposeCtxErr(err error, start time.Time) error {
	switch err {
	case context.Canceled:
		return ErrCanceled

	case context.DeadlineExceeded:
		s.leadTimeMu.RLock()
		curLeadElected := s.leadElectedTime
		s.leadTimeMu.RUnlock()
		prevLeadLost := curLeadElected.Add(-2 * time.Duration(s.Cfg.ElectionTicks) * time.Duration(s.Cfg.TickMs) * time.Millisecond)
		if start.After(prevLeadLost) && start.Before(curLeadElected) {
			return ErrTimeoutDueToLeaderFail
		}
		lead := types.ID(s.getLead())
		switch lead {
		case types.ID(raft.None):
			// TODO:返回错误以指定发生这种情况的原因是群集现在没有leader 
		case s.ID():
			if !isConnectedToQuorumSince(s.r.transport, start, s.ID(), s.cluster.Members()) {
				return ErrTimeoutDueToConnectionLost
			}
		default:
			if !isConnectedSince(s.r.transport, start, lead) {
				return ErrTimeoutDueToConnectionLost
			}
		}
		return ErrTimeout

	default:
		return err
	}
}

func (s *EtcdServer) KV() mvcc.WatchableKV { return s.kv }
func (s *EtcdServer) Backend() backend.Backend {
	s.bemu.Lock()
	defer s.bemu.Unlock()
	return s.be
}

func (s *EtcdServer) AuthStore() auth.AuthStore { return s.authStore }

func (s *EtcdServer) restoreAlarms() error {
	s.applyV3 = s.newApplierV3()
	as, err := v3alarm.NewAlarmStore(s.lg, schema.NewAlarmBackend(s.lg, s.be))
	if err != nil {
		return err
	}
	s.alarmStore = as
	if len(as.Get(pb.AlarmType_NOSPACE)) > 0 {
		s.applyV3 = newApplierV3Capped(s.applyV3)
	}
	if len(as.Get(pb.AlarmType_CORRUPT)) > 0 {
		s.applyV3 = newApplierV3Corrupt(s.applyV3)
	}
	return nil
}

// GoAttach在给定函数上创建goroutine，并使用
// etcdserver waitgroup跟踪它。
// 传递的函数应在s.StoppingNotify（）上中断。
// EtcdServer.goAttach（） 方法中会启动一个后台 goroutine 执行传入的 函数
func (s *EtcdServer) GoAttach(f func()) {
	s.wgMu.RLock() // 此阻止正在进行的关闭（s.停止）
	defer s.wgMu.RUnlock()
	select {
	case <-s.stopping:	// 检测当前 EtcdServer 实例是否已停止
		lg := s.Logger()
		lg.Warn("server has stopped; skipping GoAttach")
		return
	default:
	}

	// 调用 EtcdServer.Stop （）时，需要等待该后台 goroutine 结束之后才返回
	s.wg.Add(1)
	go func() {	// 启动一个后台 goroutine ， 执行传入的回调 函数
		defer s.wg.Done()	// 该后台 goroutine 结束时调用
		f()
	}()
}

func (s *EtcdServer) Alarms() []*pb.AlarmMember {
	return s.alarmStore.Get(pb.AlarmType_NONE)
}

// 如果本地成员是本地成员，IsLearner将返回
func (s *EtcdServer) IsLearner() bool {
	return s.cluster.IsLocalMemberLearner()
}

// 如果群集中存在具有给定id的成员，IsMemberExist将返回。
func (s *EtcdServer) IsMemberExist(id types.ID) bool {
	return s.cluster.IsMemberExist(id)
}

// raftStatus返回此etcd节点的raft状态。
func (s *EtcdServer) raftStatus() raft.Status {
	return s.r.Node.Status()
}

func (s *EtcdServer) Version() *serverversion.Manager {
	return serverversion.NewManager(s.Logger(), newServerVersionAdapter(s))
}
